package core

import (
	"kuafu/config"
	"kuafu/core/plugin"
	"kuafu/core/plugin/input"
	"kuafu/core/plugin/output"
	"kuafu/progress/log"
	"kuafu/queue"
	"time"
)

type PipeLine struct {
	inputs  []input.InPlugin
	outputs []output.OutPlugin
}

func (pipeLine *PipeLine) Start(watcher *Watcher, config *config.AppConfig) {

	buffer := queue.Create[*plugin.LogEntry](config.Queue.Size)

	//生产者
	go func() {
		for {
			select {
			case line := <-watcher.tailCh.Lines:
				var entry *plugin.LogEntry
				for _, currentPlugin := range pipeLine.inputs {
					entry = currentPlugin.Process(line)
				}
				if entry == nil {
					continue
				}
				entry.Instance = config.Instance.Name
				//过滤配置的时间之前的数据
				if config.OffSet.OffsetTime != "" {
					offSetTime, err := time.Parse("2006-01-02 15:04:05", config.OffSet.OffsetTime)
					if err != nil {
						log.Log.Errorf("配置偏移时间解析失败 :%s", config.OffSet.OffsetTime)
					}
					if offSetTime.After(time.Time(entry.LogTime)) {
						continue
					}
				}

				buffer.Put(entry)
			}
		}
	}()

	//消费者
	go func() {
		for true {
			data := buffer.Take()
			for _, currentPlugin := range pipeLine.outputs {
				currentPlugin.Consume(data)
			}
		}
	}()
}
